package com.bw.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternFlatSelectFunction;
import org.apache.flink.cep.PatternFlatTimeoutFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;

/**
 * @ProjectName: BigData
 * @Package: com.bw.app.dwm
 * @ClassName: UserJumpDetailApp
 * @Author: Gy
 * @Description: 用户跳出行为过滤
 * @Date: 2021/11/12 15:51
 */
public class UserJumpDetailApp {
    public static void main(String[] args) throws Exception {
        //todo 1.基本环境准备
        //1.1 准备本地测试流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2设置并行度
        env.setParallelism(4);

        //todo 2.从kafka中读取数据
        String sourceTopic = "dwd_page_log";
        String groupId = "user_jump_detail_group";
        String sinkTopic = "dwm_user_jump_detail";

        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getKafkaSource(sinkTopic, groupId);
        DataStreamSource<String> dataStream = env.addSource(kafkaSource);

        //todo 3.对读取到的数据进行结构的转换
        SingleOutputStreamOperator<JSONObject> jsonObjDS = dataStream.map(jsonStr -> JSON.parseObject(jsonStr));

        //todo 4.指定事件字段
        SingleOutputStreamOperator<JSONObject> jsonObjWithTSDS = jsonObjDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forMonotonousTimestamps().withTimestampAssigner(
                        new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject jsonObject, long l) {
                                return jsonObject.getLong("ts");
                            }
                        }
                )
        );

        //todo 5.按照mid分组
        KeyedStream<JSONObject, String> keyByMidDS = jsonObjWithTSDS.keyBy(
                jsonObj -> jsonObj.getJSONObject("common").getString("mid")
        );

        //todo 6.配置CEP表达式
        Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("first")
                .where(
                        //模式1：不是从其他页面跳转过来的页面，是一个首次访问页面
                        new SimpleCondition<JSONObject>() {
                            @Override
                            public boolean filter(JSONObject jsonObj) throws Exception {
                                //获取last_page_id
                                String lastPageId = jsonObj.getJSONObject("page").getString("last_page_id");
                                //判断是否为null 将为空的保留，非空的过滤掉
                                if (lastPageId == null || lastPageId.length() == 0) {
                                    return true;
                                }
                                return false;
                            }
                        }
                )
                .next("next")
                .where(
                        //模式2. 判断是否对页面做了访问
                        new SimpleCondition<JSONObject>() {
                            @Override
                            public boolean filter(JSONObject jsonObject) throws Exception {
                                //获取当前页面的id
                                String pageId = jsonObject.getJSONObject("page").getString("page_id");
                                //判断当前访问的页面id是否为null
                                if (pageId!=null && pageId.length()>0){
                                    return true;
                                }
                                return false;
                            }
                        }
                )
                //3.时间限制模式
                .within(Time.milliseconds(10000));
        //todo 7.根据：CEP表达式筛选流
        PatternStream<JSONObject> patternStream = CEP.pattern(keyByMidDS, pattern);

        //todo 8.从筛选之后的流中，提取数据  将超时数据放到测输出流
        OutputTag<String> timeoutTag = new OutputTag<String>("timeout") {};

        SingleOutputStreamOperator<String> filterDS = patternStream.flatSelect(
                timeoutTag,
                //处理超时数据
                new PatternFlatTimeoutFunction<JSONObject, String>() {
                    @Override
                    public void timeout(Map<String, List<JSONObject>> pattern, long l, Collector<String> collector) throws Exception {
                        //获取所有符合first的JSON对象
                        List<JSONObject> jsonObjectList = pattern.get("first");
                        //注意：在timeout方法中的数据都会被参数1中的标签标记
                        for (JSONObject jsonObject : jsonObjectList) {
                            collector.collect(jsonObject.toJSONString());
                        }
                    }
                },
                //处理的没有超时数据
                new PatternFlatSelectFunction<JSONObject, String>() {
                    @Override
                    public void flatSelect(Map<String, List<JSONObject>> map, Collector<String> collector) throws Exception {
                        //没有超时的数据，不在外面的统计范围之内，所以这里不需要写什么代码
                    }
                }
        );

        //todo 9.从测输出流中获取超时数据
        DataStream<String> jumpDS = filterDS.getSideOutput(timeoutTag);

        //todo 10.将跳出数据写回到kafka的DWM层
        jumpDS.addSink(MyKafkaUtil.getKafkaSink(sinkTopic));

        env.execute();
    }
}